package io.gitee.hefren.hhmq.client;

import com.google.common.base.MoreObjects;
import io.gitee.hefren.hhmq.model.HhMessage;

import static com.google.common.base.Preconditions.*;

import java.util.List;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @Date 2024/7/21
 * @Author lifei
 */
public class HhConsumer {

    private static final AtomicLong idGenerator = new AtomicLong(1);
    private String id;
    private final HhBroker hhBroker;
    private String topic;
    private HhMessageListener hhMessageListener;

    public HhConsumer(HhBroker hhBroker) {
        this.id = "hh-consumer-" + idGenerator.getAndIncrement();
        this.hhBroker = hhBroker;
    }

    public String getId() {
        return id;
    }

    public void setId(String id) {
        this.id = id;
    }
    /**
     * 订阅主题
     * @param topic
     */
    public void subscribe(String topic) {
        checkState(Objects.nonNull(topic) && !topic.isBlank(), "topic not null");
        this.topic = topic;
        hhBroker.subscribe(topic, id);
    }

    /**
     * 拉取消息
     * @return
     */
    public HhMessage<?> receive() {
        checkState(Objects.nonNull(topic) && !topic.isBlank(), "Please subscribe to topic first!");
        return hhBroker.receive(topic, id);
    }

    public List<HhMessage<?>> batchReceive(int size) {
        checkState(Objects.nonNull(topic) && !topic.isBlank(), "Please subscribe to topic first!");
        return hhBroker.batchReceive(topic, id, size);
    }

    /**
     * 取消订阅主题
     */
    public void unsubscribe() {
        checkState(Objects.nonNull(topic) && !topic.isBlank(), "Please subscribe to topic first!");
        hhBroker.unsubscribe(topic, id);
    }

    public void ack(Integer offset) {
        checkState(Objects.nonNull(topic) && !topic.isBlank(), "Please subscribe to topic first!");
        hhBroker.ack(topic, id, offset);
    }

    /**
     * 确认消费的位置
     * @param hhMessage
     */
    public void ack(HhMessage<?> hhMessage) {
        if (Objects.isNull(hhMessage)) {
            return;
        }
        checkState(Objects.nonNull(topic) && !topic.isBlank(), "Please subscribe to topic first!");
        int offset = Integer.parseInt(hhMessage.getHeaders().get(HhMessage.OFFSET_KEY));
        ack(offset);
    }

    /**
     * 添加监听
     * @param hhMessageListener
     */
    public boolean addMessageListener(HhMessageListener hhMessageListener) {
        checkState(Objects.nonNull(topic) && !topic.isBlank(), "Please subscribe to topic first!");
        this.hhMessageListener = hhMessageListener;
        return hhBroker.addConsumer(this);
    }

    public String getTopic() {
        return topic;
    }

    public HhMessageListener getHhMessageListener() {
        return hhMessageListener;
    }

    @Override
    public String toString() {
        return MoreObjects.toStringHelper(HhConsumer.class)
                .add("id", id)
                .add("topic", topic)
                .toString();
    }



}
